#include "TopicHandler.h"
#include "log.h"
#include "json.h"


int TopicHandler::ProcessPacket(CHostBase* host, SocketHandler * pHandler, NETInputPacket* inputPkg)
{
    int cmd = inputPkg->GetCmdType();
    // log_debug("cmd=0x%x,  svid=%d, ", inputPkg->GetCmdType(), pHandler->getId());
    switch (cmd)
    {
        case topic_proto::TOPIC_CMD_SUBSCRIB_REQ:
            log_debug("ProcSubscrib init");
            ProcSubscrib(host, pHandler, inputPkg);
            break;
        case topic_proto::TOPIC_CMD_UNSUBSCRIB_REQ:
            log_debug("ProcUnSubscrib init");
            ProcUnSubscrib(host, pHandler, inputPkg);
            break;
        case topic_proto::TOPIC_CMD_PUBLISH_PUSH:
            log_debug("ProcPublish init");
            ProcPublish(host, pHandler, inputPkg);
            break;
        default:
            // log_error("cmd=0x%x,  svid=%d, err no proc", cmd, pHandler->getId());
            return -1;
    }
    return 0;
}

void TopicHandler::ProcSubscrib(CHostBase* host, SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	topic_proto::SubscirbReq req;
	topic_proto::SubscirbResp resp;
	do
	{
		if (!inputPkg->parsePb(&req))
		{
			log_error("parse pb topic_proto::SubscirbReq failed");
			resp.set_err_code(1);
			break;
		}
		resp.set_trans(req.trans());

		std::int32_t svtype = pHandler->getType();
		std::int32_t svid = pHandler->getId();
		if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_RANDOM)
		{
			for (auto topic_event : req.topic_event())
			{
				AddServiceRandom(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_ID_MOD)
		{
			for (auto topic_event : req.topic_event())
			{
				AddServiceIdMod(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_UUID_MOD)
		{
			for (auto topic_event : req.topic_event())
			{
				AddServiceUuidMod(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_BROADCAST)
		{
			for (auto topic_event : req.topic_event())
			{
				AddServiceAll(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::INSTANCE)
		{
			for (auto topic_event : req.topic_event())
			{
				AddInstance(topic_event, svtype, svid);
			}
		}
		else
		{
			log_error("subscrib type no valid,%s", req.ShortDebugString().c_str());
			resp.set_err_code(1);
			break;
		}
		resp.set_err_code(0);
	} while (false);
	host->SendPb(pHandler, topic_proto::TOPIC_CMD_SUBSCRIB_RESP, &resp);
}
void TopicHandler::ProcUnSubscrib(CHostBase* host, SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	topic_proto::UnSubscirbReq req;
	topic_proto::UnSubscirbResp resp;
	do
	{
		if (!inputPkg->parsePb(&req))
		{
			log_error("parse pb topic_proto::UnSubscirbReq failed");
			resp.set_err_code(1);
			break;
		}
		resp.set_trans(req.trans());

		std::int32_t svtype = pHandler->getType();
		std::int32_t svid = pHandler->getId();
		if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_RANDOM)
		{
			for (auto topic_event : req.topic_event())
			{
				RemoveServiceRandom(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_ID_MOD)
		{
			for (auto topic_event : req.topic_event())
			{
				RemoveServiceIdMod(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_UUID_MOD)
		{
			for (auto topic_event : req.topic_event())
			{
				RemoveServiceUuidMod(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::SERVICE_BROADCAST)
		{
			for (auto topic_event : req.topic_event())
			{
				RemoveServiceAll(topic_event, svtype);
			}
		}
		else if (req.subscrib_type() == topic_proto::SubscirbType::INSTANCE)
		{
			for (auto topic_event : req.topic_event())
			{
				RemoveInstance(topic_event, svid, svtype);
			}
		}
		else
		{
			log_error("subscrib type no valid,%s", req.ShortDebugString().c_str());
			resp.set_err_code(1);
			break;
		}
		resp.set_err_code(0);
	} while (false);
	host->SendPb(pHandler, topic_proto::TOPIC_CMD_UNSUBSCRIB_RESP, &resp);
}
void TopicHandler::ProcPublish(CHostBase* host, SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	topic_proto::Topic topic;
	do
	{
		if (!inputPkg->parsePb(&topic))
		{
			log_error("parse pb topic_proto::Topic failed");
			break;
		}
		// TODO: 考虑放到队列中, 监控队列长度
		Dispatch(host, topic);
	} while (false);
}

void TopicHandler::ReloadJsonConf(const std::string& json_conf)
{
    do
    {
        Json::Value json;
        Json::Reader reader;
        if (!reader.parse(json_conf, json))
        {
            log_error("parse json failed,%s", json_conf.c_str());
            break;
        }
        if (!json.isArray())
        {
            log_error("json not array,%s", json_conf.c_str());
            break;
        }

        std::map<std::int32_t, std::set<std::int32_t>> event2serviceidmod;
        std::map<std::int32_t, std::set<std::int32_t>> event2serviceuuidmod;
        std::map<std::int32_t, std::set<std::int32_t>> event2servicerandom;
        std::map<std::int32_t, std::set<std::int32_t>> event2serviceall;
        std::map<std::int32_t, std::map<std::int32_t,std::int32_t>> event2instance;

        event2serviceidmod.swap(m_event2serviceidmod);
        event2serviceuuidmod.swap(m_event2serviceuuidmod);
        event2servicerandom.swap(m_event2servicerandom);
        event2serviceall.swap(m_event2serviceall);
        event2instance.swap(m_event2instance);

        bool finish = true;
        for (int i = 0; i < (int)json.size(); i++)
        {
            Json::Value& obj = json[i];
            std::int32_t subscrib_type = 0;
            std::int32_t topic_event = 0;
            std::int32_t service = 0;
            std::int32_t instance = 0;
            if (!obj.isMember("subscrib_type") || !obj["subscrib_type"].isInt())
            {
                finish = false; break;
            }
            subscrib_type = obj["subscrib_type"].asInt();
            if (!obj.isMember("topic_event") || !obj["topic_event"].isInt())
            {
                finish = false; break;
            }
            topic_event = obj["topic_event"].asInt();
            if (!obj.isMember("service") || !obj["service"].isInt())
            {
                finish = false; break;
            }
            service = obj["service"].asInt();
            if (subscrib_type == topic_proto::SubscirbType::INSTANCE)
            {
                if (!obj.isMember("instance") || !obj["instance"].isInt())
                {
                    finish = false; break;
                }
                instance = obj["instance"].asInt();
            }
                
            if (subscrib_type == topic_proto::SubscirbType::SERVICE_RANDOM)
            {
                AddServiceRandom(topic_event, service);
            }
            else if (subscrib_type == topic_proto::SubscirbType::SERVICE_ID_MOD)
            {
                AddServiceIdMod(topic_event, service);
            }
            else if (subscrib_type == topic_proto::SubscirbType::SERVICE_UUID_MOD)
            {
                AddServiceUuidMod(topic_event, service);
            }
            else if (subscrib_type == topic_proto::SubscirbType::SERVICE_BROADCAST)
            {
                AddServiceAll(topic_event, service);
            }
            else if (subscrib_type == topic_proto::SubscirbType::INSTANCE)
            {
                AddInstance(topic_event, service, instance);
            }
            else
            {
                log_error("subscrib type no valid,%d", subscrib_type);
                finish = false; break;
            }
        }

        if (finish)
        {
            log_info("reload success");
        }
        else
        {
            log_info("reload failed");
            // 加载失败则还原
            m_event2serviceidmod.swap(event2serviceidmod);
            m_event2serviceuuidmod.swap(event2serviceuuidmod);
            m_event2servicerandom.swap(event2servicerandom);
            m_event2serviceall.swap(event2serviceall);
            m_event2instance.swap(event2instance);
        }
    } while (false);
}

// 字符串转整数
int StrToInt(const std::string& str)
{
	int value = 0;
	std::string::const_iterator it = str.begin();
	for (; it != str.end(); it++)
	{
		value += *it;
	}
    return value;
}

void TopicHandler::Dispatch(CHostBase* host, topic_proto::Topic& topic)
{
    if (host == nullptr)
    {
		log_error("host is null, topic:%s",topic.ShortDebugString().c_str());
        return;
    }

	for (auto& service : ServiceRandomHit(topic.topic_event()))
	{
		// 随机发送一个服务实例
		log_debug("topic:%d,id:%d,uuid:%s,server:%d",topic.topic_event(), topic.id(), topic.uuid().c_str(), service);
		host->SendtoSrv(topic_proto::TOPIC_CMD_DISPATCH_PUSH, service, &topic, 0);
	}

	for (auto& service : ServiceIdModHit(topic.topic_event()))
	{
		// 按 id 取模 发送
		log_debug("topic:%d,id:%d,uuid:%s,server:%d",topic.topic_event(), topic.id(), topic.uuid().c_str(), service);
		host->SendtoSrv(topic_proto::TOPIC_CMD_DISPATCH_PUSH, service, &topic, topic.id());
	}
	
	int uuid_mod = StrToInt(topic.uuid());
	for (auto& service : ServiceUuidModHit(topic.topic_event()))
	{
		// 按 uuid 取模 发送
		log_debug("topic:%d,id:%d,uuid:%s,uuid_mod:%d,server:%d",topic.topic_event(), topic.id(), topic.uuid().c_str(), uuid_mod, service);
		host->SendtoSrv(topic_proto::TOPIC_CMD_DISPATCH_PUSH, service, &topic, uuid_mod);
	}
	
	for (auto& service : ServiceAllHit(topic.topic_event()))
	{
		// 广播消息
		log_debug("topic:%d,id:%d,uuid:%s,server:%d",topic.topic_event(), topic.id(), topic.uuid().c_str(), service);
		host->BroadcastMessageDirect(topic_proto::TOPIC_CMD_DISPATCH_PUSH, &topic, service);
	}

	for (auto& item : InstanceHit(topic.topic_event()))
	{
		// 发送到指定的服务实例
		log_debug("topic:%d,id:%d,uuid:%s,server:%d,instance:%d",topic.topic_event(), topic.id(), topic.uuid().c_str(), item.second, item.first);
		host->SendtoOneSrv(topic_proto::TOPIC_CMD_DISPATCH_PUSH, &topic, item.second, item.first);
	}
}

void TopicHandler::AddServiceIdMod(std::int32_t topic_event, std::int32_t service)
{
    m_event2serviceidmod[topic_event].emplace(service);
}
void TopicHandler::RemoveServiceIdMod(std::int32_t topic_event, std::int32_t service)
{
    m_event2serviceidmod[topic_event].erase(service);
}
const std::set<std::int32_t>& TopicHandler::ServiceIdModHit(std::int32_t topic_event)
{
    static std::set<std::int32_t> empty;
    auto it = m_event2serviceidmod.find(topic_event);
    if (it == m_event2serviceidmod.end())
    {
        return empty;
    }
    return it->second;
}

void TopicHandler::AddServiceUuidMod(std::int32_t topic_event, std::int32_t service)
{
    m_event2serviceuuidmod[topic_event].emplace(service);
}
void TopicHandler::RemoveServiceUuidMod(std::int32_t topic_event, std::int32_t service)
{
    m_event2serviceuuidmod[topic_event].erase(service);
}
const std::set<std::int32_t>& TopicHandler::ServiceUuidModHit(std::int32_t topic_event)
{
    static std::set<std::int32_t> empty;
    auto it = m_event2serviceuuidmod.find(topic_event);
    if (it == m_event2serviceuuidmod.end())
    {
        return empty;
    }
    return it->second;
}

void TopicHandler::AddServiceRandom(std::int32_t topic_event, std::int32_t service)
{
    m_event2servicerandom[topic_event].emplace(service);
}
void TopicHandler::RemoveServiceRandom(std::int32_t topic_event, std::int32_t service)
{
    m_event2servicerandom[topic_event].erase(service);
}
const std::set<std::int32_t>& TopicHandler::ServiceRandomHit(std::int32_t topic_event)
{
    static std::set<std::int32_t> empty;
    auto it = m_event2servicerandom.find(topic_event);
    if (it == m_event2servicerandom.end())
    {
        return empty;
    }
    return it->second;
}

void TopicHandler::AddServiceAll(std::int32_t topic_event, std::int32_t service)
{
    m_event2serviceall[topic_event].emplace(service);
}
void TopicHandler::RemoveServiceAll(std::int32_t topic_event, std::int32_t service)
{
    m_event2serviceall[topic_event].erase(service);
}
const std::set<std::int32_t>& TopicHandler::ServiceAllHit(std::int32_t topic_event)
{
    static std::set<std::int32_t> empty;
    auto it = m_event2serviceall.find(topic_event);
    if (it == m_event2serviceall.end())
    {
        return empty;
    }
    return it->second;
}

void TopicHandler::AddInstance(std::int32_t topic_event, std::int32_t service, std::int32_t instance)
{
    m_event2instance[topic_event].emplace(instance, service);
}
void TopicHandler::RemoveInstance(std::int32_t topic_event, std::int32_t service, std::int32_t instance)
{
    m_event2instance[topic_event].erase(instance);
}
const std::map<std::int32_t,std::int32_t>& TopicHandler::InstanceHit(std::int32_t topic_event)
{
    static std::map<std::int32_t,std::int32_t> empty;
    auto it = m_event2instance.find(topic_event);
    if (it == m_event2instance.end())
    {
        return empty;
    }
    return it->second;

}




















